package beautiful.butterfly.schedule.task_manager;

import beautiful.butterfly.schedule.MultipleTaskDeal;
import beautiful.butterfly.schedule.SingleTaskDeal;
import beautiful.butterfly.schedule.TaskDeal;
import beautiful.butterfly.schedule.TaskItemAndParameter;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Array;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

public class SleepProcessor<Task> implements IProcessor, Runnable {

    private static transient Logger logger = LoggerFactory.getLogger(SleepProcessor.class);
    final Lock lock = new Lock();
    @Setter
    @Getter
    protected AbstractScheduleManager abstractScheduleManager;
    @Setter
    @Getter
    protected TaskDeal<Task> taskDeal;
    @Setter
    @Getter
    protected List<Task> taskList = new CopyOnWriteArrayList<Task>();
    @Setter
    @Getter
    List<Thread> threadList = new CopyOnWriteArrayList<Thread>();
    @Setter
    @Getter
    TaskType taskType;
    @Setter
    @Getter
    boolean isMutilTask = false;

    @Setter
    @Getter
    boolean isStopSchedule = false;
    @Setter
    @Getter
    boolean isSleeping = false;
    @Setter
    @Getter
    Statistics statistics;


    public SleepProcessor(AbstractScheduleManager abstractScheduleManager,
                          TaskDeal<Task> taskDeal, Statistics statistics) throws Exception {
        this.abstractScheduleManager = abstractScheduleManager;
        this.statistics = statistics;
        this.taskType = this.abstractScheduleManager.getTaskType();
        this.taskDeal = taskDeal;
        if (this.taskDeal instanceof SingleTaskDeal<?>) {
            if (taskType.getMaxExecuteNumber() > 1) {
                taskType.setMaxExecuteNumber(1);//自动修正
            }
            isMutilTask = false;
        } else {
            isMutilTask = true;
        }
        if (taskType.getFetchDataNumber() >= taskType.getThreadNumber() * 10) {
            logger.warn("fetchnum >= threadnum*10 ");
        }
        for (int i = 0; i < taskType.getThreadNumber(); i++) {
            this.startThread(i);
        }
    }


    public void stopSchedule() throws Exception {

        this.isStopSchedule = true;

        this.taskList.clear();
    }

    private void startThread(int index) {
        Thread thread = new Thread(this);
        threadList.add(thread);
        String threadName = this.abstractScheduleManager.getTaskHandler().getTaskType() + "-"
                + this.abstractScheduleManager.getCurrentSerialNumber() + "-exe"
                + index;
        thread.setName(threadName);
        thread.start();
    }

    public synchronized Object getTask() {
        if (this.taskList.size() > 0)
            return this.taskList.remove(0);
        return null;
    }

    public synchronized Object[] getTasks() {
        if (this.taskList.size() == 0) {
            return null;
        }
        int size = taskList.size() > taskType.getMaxExecuteNumber() ? //min
                taskType.getMaxExecuteNumber()//
                : taskList.size();//

        Object[] taskList = null;
        if (size > 0) {
            taskList = (Object[]) Array.newInstance(this.taskList.get(0).getClass(), size);
        }
        for (int i = 0; i < size; i++) {
            taskList[i] = this.taskList.remove(0);
        }
        return taskList;
    }

    public void clearTaskList() {
        this.taskList.clear();
    }

    public boolean taskListIsAllFinshed() {
        return this.taskList.size() == 0;
    }

    public boolean isSleeping() {
        return this.isSleeping;
    }

    protected int getTaskList() {
        try {
            if (this.taskType.getSleepTimeInterval() > 0) {

                this.isSleeping = true;
                Thread.sleep(taskType.getSleepTimeInterval());
                this.isSleeping = false;

            }

            List<TaskItemAndParameter> taskItemAndParameterList = this.abstractScheduleManager.getCurrentTaskItemAndParameterList();
            if (taskItemAndParameterList.size() > 0) {
                List<TaskItemAndParameter> newTaskItemAndParameterList = new ArrayList<TaskItemAndParameter>();
                synchronized (taskItemAndParameterList) {
                    for (TaskItemAndParameter taskItemAndParameter : taskItemAndParameterList) {
                        newTaskItemAndParameterList.add(taskItemAndParameter);
                    }
                }
                List<Task> taskList = this.taskDeal.getTaskList(//
                        taskType.getTaskParameter(),//
                        abstractScheduleManager.getTaskHandler().getOwnSign(),//
                        this.abstractScheduleManager.getTaskItemCount(), //
                        taskType.getFetchDataNumber(), newTaskItemAndParameterList//
                );//
                abstractScheduleManager.getTaskHandler().setLastFetchDataTime(new Timestamp(abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()));
                if (taskList != null) {
                    this.taskList.addAll(taskList);
                }
            } else {
            }
            addFetchNum(taskList.size());
            return this.taskList.size();
        } catch (Throwable ex) {
            logger.error("Get tasks error.", ex);
        }
        return 0;
    }

    @SuppressWarnings({"rawtypes", "unchecked", "static-access"})
    public void run() {
        try {
            long startTime = 0;
            while (true) {
                this.lock.addWaitThread();
                Object task;
                while (true) {
                    if (this.isStopSchedule == true) {
                        this.lock.realseThread();
                        this.lock.notifyOtherWaitThread();
                        synchronized (this.threadList) {
                            this.threadList.remove(Thread.currentThread());
                            if (this.threadList.size() == 0) {
                                this.abstractScheduleManager.unRegisterTaskHandler();
                            }//all over
                        }
                        return;
                    }

                    if (this.isMutilTask == false) {
                        task = this.getTask();
                    } else {
                        task = this.getTasks();
                    }

                    if (task == null) {
                        break;
                    }

                    try {
                        startTime = abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis();
                        if (this.isMutilTask == false) {
                            if (((SingleTaskDeal) this.taskDeal).execute(task, abstractScheduleManager.getTaskHandler().getOwnSign()) == true) {
                                addSuccessNum(1, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                        - startTime);
                            } else {
                                addFailNum(1, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                        - startTime);
                            }
                        } else {
                            if (((MultipleTaskDeal) this.taskDeal)
                                    .execute((Object[]) task, abstractScheduleManager.getTaskHandler().getOwnSign()) == true) {
                                addSuccessNum(((Object[]) task).length, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                        - startTime);
                            } else {
                                addFailNum(((Object[]) task).length, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                        - startTime);
                            }
                        }
                    } catch (Throwable ex) {
                        if (this.isMutilTask == false) {
                            addFailNum(1, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis() - startTime);
                        } else {
                            addFailNum(((Object[]) task).length, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                    - startTime);
                        }
                        logger.warn("Task :" + task + " ", ex);
                    }
                }

                if (this.lock.realseThreadAndReturnIsLastThread() == false) {
                    int size = 0;
                    Thread.currentThread().sleep(100);
                    startTime = abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis();


                    size = this.getTaskListSize();
                    if (size > 0) {
                        this.lock.notifyOtherWaitThread();
                    } else {
                        if (this.isStopSchedule == false && this.abstractScheduleManager.isContinueWhenData() == true) {
                            this.isSleeping = true;
                            Thread.currentThread().sleep(this.abstractScheduleManager.getTaskType().getSleepTimeNoData());
                            this.isSleeping = false;

                        } else {
                            this.lock.notifyOtherWaitThread();
                        }
                    }
                    this.lock.realseThread();
                } else {
                    this.lock.currentThreadWait();
                }
            }
        } catch (Throwable e) {
            logger.error(e.getMessage(), e);
        }
    }

    public void addFetchNum(long num) {

        this.statistics.addFetchDataCount(1);
        this.statistics.addFetchDataNum(num);
    }

    public void addSuccessNum(long num, long spendTime) {
        this.statistics.addDealDataSucess(num);
        this.statistics.addDealSpendTime(spendTime);
    }

    public void addFailNum(long num, long spendTime) {
        this.statistics.addDealDataFail(num);
        this.statistics.addDealSpendTime(spendTime);
    }
}
